package org.xq.softcup.mq.admin.broker;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.xq.softcup.mq.admin.storage.MqStorageManager;
import org.xq.softcup.mq.admin.model.RegData;
import org.xq.softcup.mq.admin.model.MqTopic;
import org.xq.softcup.mq.admin.dao.IMqMsgDao;
import org.xq.softcup.mq.admin.dao.IMqTopicDao;
import org.xq.softcup.mq.admin.service.impl.MqRegServiceImpl;
import org.xq.softcup.mq.client.broker.IMqBroker;
import org.xq.softcup.mq.client.message.MqMessage;
import org.xq.softcup.mq.client.message.MqMessageStatus;
import org.xq.softcup.mq.remoting.net.NetEnum;
import org.xq.softcup.mq.remoting.provider.XQRpcProviderFactory;
import org.xq.softcup.mq.serialize.Serializer;
import org.xq.softcup.mq.util.IpUtil;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Broker消息处理服务实现类
 */
@Slf4j
@Component
public class MqBrokerImpl implements IMqBroker, InitializingBean, DisposableBean {

    @Value("${mq.rpc.remoting.ip}")
    private String ip;

    @Value("${mq.rpc.remoting.port}")
    private int port;

    @Value("${mq.log.logretentiondays}")
    private int logretentiondays;

    @Resource
    private IMqMsgDao mqMessageDao;
//    @Resource
//    private IMqTopicService mqTopicService;
//    @Resource
//    private IMqTopicDao mqTopicDao;

    private AtomicInteger count = new AtomicInteger(0);
    @Resource
    private MqStorageManager mqStorageManager;

    @Override
    public void afterPropertiesSet() throws Exception {
        // 初始化服务
        initServer();
        // 初始化线程
        initThead();
    }

    @Override
    public void destroy() throws Exception {
        // 销毁服务器
        destoryServer();
        // 销毁线程
        destroyThread();
    }
    //新消息队列
    private LinkedBlockingQueue<MqMessage> newMessageQueue = new LinkedBlockingQueue<MqMessage>();
    //回调消息队列
    private LinkedBlockingQueue<MqMessage> callbackMessageQueue = new LinkedBlockingQueue<MqMessage>();
    //等待确认的消息
    private ConcurrentHashMap<String, MqMessage> waitMsgQueue = new ConcurrentHashMap<>();
    //创建可缓存的线程池
    private ExecutorService executorService = Executors.newCachedThreadPool();
    private volatile boolean executorStoped = false; //多线程控制器

    public void initThead() throws Exception {

        /**
         * Broker异步存储消息, 多线程
         */
        for (int i = 0; i < 3; i++) {
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    while (!executorStoped) {
                        try {
                            MqMessage message = newMessageQueue.take();
                            if (message != null) {
                                // 加载
                                List<MqMessage> messageList = new ArrayList<>();
                                messageList.add(message);
                                //获取剩余的，最多获取50个数据
                                List<MqMessage> otherMessageList = new ArrayList<>();
                                int drainToNum = newMessageQueue.drainTo(otherMessageList, 49);
                                if (drainToNum > 0) {
                                    messageList.addAll(otherMessageList);
                                }
                                // 保存消息到数据库
//                                mqMessageDao.save(messageList);
                                // 保存到本地文件
                                mqStorageManager.getMqStorage().pushMessages(message.getTopic(), messageList);
                            }
                        } catch (Exception e) {
                            if (!executorStoped) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    }

                    // 结尾消息保存
                    List<MqMessage> otherMessageList = new ArrayList<>();
                    int drainToNum = newMessageQueue.drainTo(otherMessageList);
                    if (drainToNum> 0) {
//                        mqMessageDao.save(otherMessageList);
                        // 保存到本地文件
                        mqStorageManager.getMqStorage().pushMessages(otherMessageList.get(0).getTopic(), otherMessageList);
                    }

                }
            });
        }

        /**
         * 多线程异步回调消息
         */
        for (int i = 0; i < 3; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    while (!executorStoped) {
                        try {
                            MqMessage message = callbackMessageQueue.take();
                            if (message!=null) {
                                // 加载
                                List<MqMessage> messageList = new ArrayList<>();
                                messageList.add(message);

                                List<MqMessage> otherMessageList = new ArrayList<>();
                                int drainToNum = callbackMessageQueue.drainTo(otherMessageList, 100);
                                if (drainToNum > 0) {
                                    messageList.addAll(otherMessageList);
                                }
                                for (MqMessage mqMessage : messageList) {
                                    //核销等待消息
                                    MqMessage remove = waitMsgQueue.remove(mqMessage.getId());
                                    if (remove != null) {
                                        log.info("ACK消息数量：{}" , count.addAndGet(1));
                                    }
                                }
                                // 保存
                            }

                        } catch (Exception e) {
                            if (!executorStoped) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    }

                    // 末尾消息保存
                    List<MqMessage> otherMessageList = new ArrayList<>();
                    int drainToNum = callbackMessageQueue.drainTo(otherMessageList);
                    if (drainToNum > 0) {
                        //更新消息状态
//                        mqMessageDao.updateStatus(otherMessageList);
                        for (MqMessage mqMessage : otherMessageList) {
                            //核销等待消息
                            MqMessage remove = waitMsgQueue.remove(mqMessage.getId());
                            if (remove != null) {
                                log.info("核销消息数量：{}" , count.addAndGet(1));
                            }
                        }
                    }

                }
            });
        }


        /**
         *  自定重试消息，重试次数减1，并更新状态，重试时间每分钟
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                long sleep = 200;
                while (!executorStoped) {
                    try {
                        // 多线程重试消息
//                        String appendLog = LogUtil.makeLog("失败重试", "状态自动还原,剩余重试次数减一");
//                        int count = mqMessageDao.updateRetryCount(MqMessageStatus.FAIL.name(), MqMessageStatus.NEW.name(), appendLog);
                        if ( waitMsgQueue.size() > 0) {
                            if (newMessageQueue.size() > 0) {
                                sleep = 10;
                                continue;
                            }
                            log.info("重试等待消息, count:{}", waitMsgQueue.size());
                            List retryList = new ArrayList();
                            Set<String> keys = waitMsgQueue.keySet();
                            Iterator<String> iterator = keys.iterator();
                            while(iterator.hasNext()) {
                                String key = iterator.next();
                                MqMessage mqMessage = waitMsgQueue.remove(key); //移除等待队列，重发消息
                                retryList.add(mqMessage);
                                log.info("重试发送消息, data:{}", mqMessage.getData());
                            }
                            //重试失败的消息
                            addMessages(retryList);
                            log.info("重试消息 Size: {}", retryList.size());
                        }
                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                    try {
                        // sleep
                        TimeUnit.SECONDS.sleep(sleep);
                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });

        /**
         * 自动清除成功的消息
        if (logretentiondays >= 3) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    while (!executorStoped) {
                        try {
                            int count = mqMessageDao.cleanSuccessMessage(MqMessageStatus.SUCCESS.name(), logretentiondays);
                            log.info("清除成功的消息, 数量:{}", count);
                        } catch (Exception e) {
                            if (!executorStoped) {
                                log.error(e.getMessage(), e);
                            }
                        }
                        try {
                            TimeUnit.DAYS.sleep(logretentiondays);
                        } catch (Exception e) {
                            if (!executorStoped) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    }
                }
            });
        }
         */

        /**
         * 自动查找新的topic
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {
                        // 查找新主题, 添加信息
                        List<String> topicList = mqMessageDao.findNewTopicList();
                        if (topicList!=null && topicList.size()>0) {
                            for (String topic:topicList) {
                                MqTopic newTopic = new MqTopic();
                                newTopic.setTopic(topic);
                                mqTopicService.add(newTopic);
                            }
                        }

                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                    try {
                        TimeUnit.MINUTES.sleep(1); //睡眠1分钟
                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });
         */

    }
    //销毁线程池
    public void destroyThread(){
        executorStoped = true;
        executorService.shutdownNow();
    }
    //Broker 服务工厂类
    private XQRpcProviderFactory providerFactory;
    //初始化服务
    public void initServer() throws Exception {
        // 注册地址
        ip = (ip!=null&&ip.trim().length()>0)?ip:IpUtil.getIp();
        String address = IpUtil.getIpPort(ip, port);

        RegData regData = new RegData();
        regData.setKey(IMqBroker.class.getName());
        regData.setValue(address);
        MqRegServiceImpl.staticRegData = regData;

        // Brocker初始化服务
        providerFactory = new XQRpcProviderFactory();
        providerFactory.initConfig(NetEnum.NETTY, Serializer.SerializeEnum.HESSIAN.getSerializer(),
                ip, port, null, null, null);

        // 添加服务
        providerFactory.addService(IMqBroker.class.getName(), null, this);

        // 启动服务器
        providerFactory.start();
    }
    // 停止服务
    public void destoryServer() throws Exception {
        if (providerFactory != null) {
            providerFactory.stop();
        }
    }

    //添加新消息
    @Override
    public int addMessages(List<MqMessage> messages) {
        newMessageQueue.addAll(messages);
        return messages.size();
    }
    //消费者获取新消息
    @Override
    public List<MqMessage> pullNewMessage(String topic, String group, int consumerRank, int consumerTotal, int pagesize) {
//        List<MqMessage> list = mqMessageDao.pullNewMessage(MqMessageStatus.NEW.name(), topic, group, consumerRank, consumerTotal, pagesize);
        String msg = mqStorageManager.getMqStorage().popMessage(topic, group);
        if (!StringUtils.isEmpty(msg)) {
            //转化集合
            List<MqMessage> msgList = new ArrayList<>();
            try {
                if (msg.startsWith("[")) {
                    //转化集合
                    msgList = JSON.parseArray(msg, MqMessage.class);
                } else {
                    MqMessage mqMessage = JSON.parseObject(msg, MqMessage.class);
                    msgList = Arrays.asList(mqMessage);
                }
                addWaitMsg(msgList);
            } catch (Exception e) {
                log.error("错误：" + msg);
            }
            return msgList;
        }
        return null;
    }

    public void addWaitMsg(List<MqMessage> waitingMsgList) {
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                //放入等待队列中
                for (MqMessage mqMessage : waitingMsgList) {
                    waitMsgQueue.put(mqMessage.getId(), mqMessage);
                }
            }
        });
    }
    //锁住消息
    @Override
    public int lockMessage(long id, String appendLog) {
//        return mqMessageDao.lockMessage(id, appendLog, MqMessageStatus.NEW.name(), MqMessageStatus.RUNNING.name());
        return 0;
    }
    //回调消息
    @Override
    public int callbackMessages(List<MqMessage> messages) {
        callbackMessageQueue.addAll(messages);
        return messages.size();
    }

}
